package ex.tableapi;

import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.concat;

public class ExCureColumns extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExCureColumns demo = new ExCureColumns();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table result = demo.exCureColumns(orders);
        Table result1 = demo.exAs(orders);
        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    /**
     * 操作内存表字段
     * @param orders
     * @return
     */
    private Table exCureColumns(Table orders) {
        Table result1 = orders.addColumns(concat($("c"), "sunny"));
        Table result2 = orders.addOrReplaceColumns(concat($("c"), "sunny").as("desc"));
        Table result3 = orders.dropColumns($("b"), $("c"));
        Table result4 = orders.renameColumns($("b").as("b2"), $("c").as("c2"));
        return result4;
    }

    private Table exAs(Table orders) {
        Table result1 = orders.where($("b").isEqual(500));
        Table result2 = orders.filter($("b").isEqual(500));
        return result2;
    }

}
